/*
 * Licensed to the Tai Ping Jin Ke
 *
 * Copyright (c) 2022 .
 * All rights reserved.
 * 项目名称：太平金科-通用能力封装层-公共组件
 * 版权说明：本软件属太平金融科技服务(上海)有限公司所有，在未获得太平金融科技服务(上海)有限公司正式授权情况下，任何企业和个人，不能获取、阅读、安装、传播本软件涉及的任何受知识产权保护的内容。
 */
package com.taipingframework.utility.thread;

import com.taipingframework.utility.exception.UtilityException;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.FutureTask;
import java.util.concurrent.Semaphore;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * 并行处理封装类
 * <p>
 * （1）CountDownLatch 计数器
 * <p>
 * （2）CyclicBarrier 循环障碍
 * <p>
 * （3）Semaphore 信号量
 * <p>
 * （4）ExecutorService.invokeAll(Collection<? extends Callable<T>>) 线程池批量提交任务
 * <p>
 * （5）JDK8 新引入 CompletableFuture，支持异步无阻塞式的编程模式。
 * Future<T> + Callable<T> 模式缺点：(是指通过注解@Async以代理的方式实现的异步方法)
 * - 获取结果需阻塞；
 * - 无法手动完成
 * - 多个Future不能链接在一起
 * - 不能将多个Future组合在一起
 * - Future 无异常处理
 * - CompletableFuture 的出现解决了Future在实践应用中的各种短板。
 * <p>
 * （6）Future 如何获取 Callable<T> 的返回结果 Future<T>
 * <p>
 * （7）Future 如何获取 Runnable 的返回结果 Future<Void>
 * <p>
 * （8）相较于`Future<T> + Callable<T> `模式，CompletableFuture异步无阻塞式的编程模式使异步方法在实践中更加灵活、高效、便利。
 * 另外，在异步无阻塞式的编程模式下，reactor框架成为完美的替代方案，并且，基于SpringBoot构建的Servlet服务也对其提供了支持。
 */
@Slf4j
public class ParallelHelper {

    /**
     * 这里使用JDK的线程池
     */
    private static ExecutorService threadPool;

    public static void setThreadPool(ExecutorService threadPool) {
        ParallelHelper.threadPool = threadPool;
    }

    /**
     * 【并行消费】
     * 将一个大任务拆分为若干个小任务之后使用线程池并行处理。
     *
     * @param taskList 列表任务
     * @param consumer 处理列表任务的具体逻辑
     * @param <T>      列表任务的数据类型
     */
    public static <T> void parallelHandle(final List<T> taskList, Consumer<T> consumer) throws UtilityException {
        CountDownLatch counter = new CountDownLatch(taskList.size());

        Throwable[] throwableArray = new Throwable[1];
        for (final T task : taskList) {
            threadPool.execute(() -> {
                try {
                    consumer.accept(task);
                } catch (Exception ex) {
                    throwableArray[0] = ex;
                } finally {
                    counter.countDown();
                }
            });
        }

        try {
            counter.await();
            if (Objects.nonNull(throwableArray[0])) {
                throw new UtilityException(throwableArray[0]);
            }
        } catch (Exception ex) {
            throw new UtilityException(ex);
        }
    }

    /**
     * 【并行转换】
     * 将一个大任务拆分为若干个小任务之后使用线程池并行处理。
     *
     * @param taskList  列表任务
     * @param converter 处理列表任务的具体逻辑
     * @param <T>       列表任务的数据类型
     * @return 转换后的结果集
     */
    public static <T, R> List<R> parallelConvert(final List<T> taskList, Function<T, R> converter) throws UtilityException {
        CountDownLatch counter = new CountDownLatch(taskList.size());

        Throwable[] throwableArray = new Throwable[1];
        List<R> resultList = new CopyOnWriteArrayList<>();
        for (final T task : taskList) {
            threadPool.execute(() -> {
                try {
                    resultList.add(converter.apply(task));
                } catch (Exception ex) {
                    throwableArray[0] = ex;
                } finally {
                    counter.countDown();
                }
            });
        }

        try {
            counter.await();
            if (Objects.nonNull(throwableArray[0])) {
                throw new UtilityException(throwableArray[0]);
            }
        } catch (Exception ex) {
            throw new UtilityException(ex);
        }

        return resultList;
    }

    /**
     * 异步批量执行任务，调用线程被阻塞
     */
    public static <T> List<T> parallelInvoke(List<CompletableFuture<T>> taskList) throws UtilityException {
        CountDownLatch counter = new CountDownLatch(taskList.size());

        Throwable[] throwableArray = new Throwable[1];
        List<T> listResult = new CopyOnWriteArrayList<>();
        for (CompletableFuture<T> task : taskList) {
            threadPool.execute(() -> {
                try {
                    listResult.add(task.get());
                } catch (Exception ex) {
                    throwableArray[0] = ex;
                } finally {
                    counter.countDown();
                }
            });
        }

        try {
            counter.await();
            if (Objects.nonNull(throwableArray[0])) {
                throw new UtilityException(throwableArray[0]);
            }
        } catch (Exception ex) {
            throw new UtilityException(ex);
        }

        return listResult;
    }

    /**
     * 【合并汇总】
     * 将若干个小任务执行的结果进行汇总并返回，CyclicBarrier可以用于多线程计算数据，最后合并计算结果的应用场景。
     *
     * @param taskList 分多个线程处理数据
     * @param consumer 报告汇总的结果和异常
     * @param <R>      返回类型
     */
    public static <R> void parallelMerge(final List<CompletableFuture<R>> taskList, BiConsumer<List<R>, Throwable> consumer) {
        List<R> total = new CopyOnWriteArrayList<>();
        Throwable[] throwableArray = new Throwable[1];

        CyclicBarrier barrier = new CyclicBarrier(taskList.size(), () -> {
            if (Objects.isNull(throwableArray[0])) {
                consumer.accept(total, null);
            } else {
                total.clear();
                consumer.accept(total, throwableArray[0]);
            }
        });

        for (CompletableFuture<R> task : taskList) {
            threadPool.execute(() -> {
                try {
                    task.thenAccept(total::add).get();
                } catch (Exception e) {
                    throwableArray[0] = e;
                } finally {
                    try {
                        barrier.await();
                    } catch (InterruptedException | BrokenBarrierException ignored) {
                    }
                }
            });
        }
    }

    /**
     * 信号量 Semaphore 两个应用场景：
     * (1) 用于多个共享资源的互斥使用
     * (2) 用于并发线程数量的控制
     *
     * @param semaphore 信号量，控制并发数
     * @param consumer  任务
     */
    public static void permitsLimit(@NonNull Semaphore semaphore, Consumer<Integer> consumer) throws UtilityException {
        try {
            // 申请资源，如果当前已占用资源数达到`permits`则会阻塞等待
            semaphore.acquire();

            log.debug("{} 开始执行任务...", Thread.currentThread().getName());
            consumer.accept(semaphore.availablePermits());
            log.debug("{} 结束执行任务!!!", Thread.currentThread().getName());
        } catch (Exception ex) {
            throw new UtilityException(ex);
        } finally {
            semaphore.release();
        }
    }

    /**
     * 信号量 Semaphore 两个应用场景：
     * (1) 用于多个共享资源的互斥使用
     * (2) 用于并发线程数量的控制
     *
     * @param semaphore 信号量，控制并发数
     * @param task      任务
     */
    public static <V> V permitsLimit(@NonNull Semaphore semaphore, Supplier<V> task) throws UtilityException {
        try {
            // 申请资源，如果当前已占用资源数达到`permits`则会阻塞等待
            semaphore.acquire();

            log.debug("{} 开始执行任务...", Thread.currentThread().getName());
            V v = task.get();
            log.debug("{} 结束执行任务!!!", Thread.currentThread().getName());

            return v;
        } catch (Exception ex) {
            throw new UtilityException(ex);
        } finally {
            semaphore.release();
        }
    }

    /**
     * 异步执行任务，有返回值
     */
    public static <T> CompletableFuture<T> asyncInvoke(Supplier<T> supplier) {
        return CompletableFuture.supplyAsync(supplier, threadPool);
    }

    /**
     * 异步执行任务，无返回值
     */
    public static CompletableFuture<Void> asyncInvoke(Runnable runnable) {
        return CompletableFuture.runAsync(runnable, threadPool);
    }

    /**
     * 异步执行，有返回值
     * <p>
     * 用法：
     * // 误区：在主线程直接调用FutureTask<T>对象的get方法，将永远阻塞在此处。
     * FutureTask<Integer> futureTask1 = ParallelHelper.convert(() -> 12);
     * new Thread(futureTask1).start();
     * log.info(futureTask1.get()); // 阻塞，直到线程执行完成
     */
    public static <T> FutureTask<T> convert(Callable<T> task) {
        return new FutureTask<>(task);
    }

    /**
     * 异步执行，无返回值
     * <p>
     * 用法：
     * // 误区：在主线程直接调用FutureTask<Void>对象的run方法，将永远阻塞在此处。
     * FutureTask<Void> futureTask2 = ParallelHelper.convert(() -> log.info("异步请求测试"));
     * new Thread(futureTask2).start();
     * futureTask2.run(); // 阻塞，直到线程执行完成
     */
    public static FutureTask<Void> convert(Runnable task) {
        return new FutureTask<>(task, null);
    }

    /*
     * JDK 1.8 新增加的函数接口：
     *
     * 1、BiConsumer<T, U>
     * 代表了一个接受两个输入参数的操作，并且不返回任何结果
     *
     * 2、BiFunction<T, U, R>
     * 代表了一个接受两个输入参数的方法，并且返回一个结果
     *
     * 3、BinaryOperator<T>
     * 代表了一个作用于于两个同类型操作符的操作，并且返回了操作符同类型的结果
     *
     * 4、BiPredicate<T, U>
     * 代表了一个两个参数的boolean值方法
     *
     * 5、BooleanSupplier
     * 代表了boolean值结果的提供方
     *
     * 6、Consumer<T>
     * 代表了接受一个输入参数并且无返回的操作
     *
     * 7、DoubleBinaryOperator
     * 代表了作用于两个double值操作符的操作，并且返回了一个double值的结果。
     *
     * 8、DoubleConsumer
     * 代表一个接受double值参数的操作，并且不返回结果。
     *
     * 9、DoubleFunction<R>
     * 代表接受一个double值参数的方法，并且返回结果
     *
     * 10、DoublePredicate
     * 代表一个拥有double值参数的boolean值方法
     *
     * 11、DoubleSupplier
     * 代表一个double值结构的提供方
     *
     * 12、DoubleToIntFunction
     * 接受一个double类型输入，返回一个int类型结果。
     *
     * 13、DoubleToLongFunction
     * 接受一个double类型输入，返回一个long类型结果
     *
     * 14、DoubleUnaryOperator
     * 接受一个参数同为类型double, 返回值类型也为double 。
     *
     * 15、Function<T, R>
     * 接受一个输入参数，返回一个结果。
     *
     * 16、IntBinaryOperator
     * 接受两个参数同为类型int, 返回值类型也为int 。
     *
     * 17、IntConsumer
     * 接受一个int类型的输入参数，无返回值 。
     *
     * 18、IntFunction<R>
     * 接受一个int类型输入参数，返回一个结果 。
     *
     * 19、IntPredicate
     * ：接受一个int输入参数，返回一个布尔值的结果。
     *
     * 20、IntSupplier
     * 无参数，返回一个int类型结果。
     *
     * 21、IntToDoubleFunction
     * 接受一个int类型输入，返回一个double类型结果 。
     *
     * 22、IntToLongFunction
     * 接受一个int类型输入，返回一个long类型结果。
     *
     * 23、IntUnaryOperator
     * 接受一个参数同为类型int, 返回值类型也为int 。
     *
     * 24、LongBinaryOperator
     * 接受两个参数同为类型long, 返回值类型也为long。
     *
     * 25、LongConsumer
     * 接受一个long类型的输入参数，无返回值。
     *
     * 26、LongFunction<R>
     * 接受一个long类型输入参数，返回一个结果。
     *
     * 27、LongPredicate
     * R接受一个long输入参数，返回一个布尔值类型结果。
     *
     * 28、LongSupplier
     * 无参数，返回一个结果long类型的值。
     *
     * 29、LongToDoubleFunction
     * 接受一个long类型输入，返回一个double类型结果。
     *
     * 30、LongToIntFunction
     * 接受一个long类型输入，返回一个int类型结果。
     *
     * 31、LongUnaryOperator
     * 接受一个参数同为类型long, 返回值类型也为long。
     *
     * 32、ObjDoubleConsumer<T>
     * 接受一个object类型和一个double类型的输入参数，无返回值。
     *
     * 33、ObjIntConsumer<T>
     * 接受一个object类型和一个int类型的输入参数，无返回值。
     *
     * 34、ObjLongConsumer<T>
     * 接受一个object类型和一个long类型的输入参数，无返回值。
     *
     * 35、Predicate<T>
     * 接受一个输入参数，返回一个布尔值结果。
     *
     * 36、Supplier<T>
     * 无参数，返回一个结果。
     *
     * 37、ToDoubleBiFunction<T, U>
     * 接受两个输入参数，返回一个double类型结果
     *
     * 38、ToDoubleFunction<T>
     * 接受一个输入参数，返回一个double类型结果
     *
     * 39、ToIntBiFunction<T, U>
     * 接受两个输入参数，返回一个int类型结果。
     *
     * 40、ToIntFunction<T>
     * 接受一个输入参数，返回一个int类型结果。
     *
     * 41、ToLongBiFunction<T, U>
     * 接受两个输入参数，返回一个long类型结果。
     *
     * 42、ToLongFunction<T>
     * 接受一个输入参数，返回一个long类型结果。
     *
     * 43、UnaryOperator<T>
     * 接受一个参数为类型T, 返回值类型也为T。
     */
}
